package com.gitee.xhs.window.triggers;

import com.gitee.xhs.OperatorSourceFactory;
import com.gitee.xhs.customFunction.OperatorKeyBy;
import com.gitee.xhs.customWaterMark.CustomWaterStrategy;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import com.gitee.xhs.window.windowOperator.DefaultAggregateWindowFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author BigHead
 * @date 2022/7/26 14:25
 * @description SimpleWindowTriggerMain
 */
public class SimpleWindowTriggerMain {

    private final StreamExecutionEnvironment streamExecutionEnvironment;

    public SimpleWindowTriggerMain(StreamExecutionEnvironment streamExecutionEnvironment) {
        this.streamExecutionEnvironment = streamExecutionEnvironment;
    }

    public void triggerLogic() {

        KeyedStream<Transaction, Long> windowKeyedTransactionStream = streamExecutionEnvironment
                .addSource(
                        OperatorSourceFactory.unBoundLoopOneToTenOneSecondSource()
                )
                .assignTimestampsAndWatermarks(
                        new CustomWaterStrategy()
                )
                .keyBy(
                        new OperatorKeyBy()
                );

        // onElement
        onElement(windowKeyedTransactionStream);
        // onProcessingTime
        onProcessingTime(windowKeyedTransactionStream);
        // onEventTime
        onEventTime(windowKeyedTransactionStream);

    }

    private void onElement(KeyedStream<Transaction, Long> windowKeyedTransactionStream){
        windowKeyedTransactionStream
                .window(
                        TumblingEventTimeWindows.of(Time.minutes(1))
                )
                .trigger(
                        new CustomEachElementOnElementTrigger()
                )
                .aggregate(
                        new DefaultAggregateWindowFunction()
                )
                .print("自定义Trigger OnElement 效果");
    }

    private void onProcessingTime(KeyedStream<Transaction, Long> windowKeyedTransactionStream){
        windowKeyedTransactionStream
                .window(
                        TumblingEventTimeWindows.of(Time.minutes(1))
                )
                .trigger(
                        new CustomOnProcessingTimeTrigger()
                )
                .aggregate(
                        new DefaultAggregateWindowFunction()
                )
                .print("自定义Trigger OnProcessingTime 效果");
    }

    private void onEventTime(KeyedStream<Transaction, Long> windowKeyedTransactionStream){
        windowKeyedTransactionStream
                .window(
                        TumblingEventTimeWindows.of(Time.minutes(1))
                )
                .trigger(
                        new CustomOnEventTimeTrigger()
                )
                .aggregate(
                        new DefaultAggregateWindowFunction()
                )
                .print("自定义Trigger OnEventTime 效果");
    }

}
